/*
 * Package il.ac.biu.cs.grossmm.impl.activeData
 * File DataLayer.java
 * 
 * This is a part of presence service framework prototype. 
 * See javadoc for more information.
 *  
 * Copyright (C) 2006 Grossmann Mark
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public License
 * as published by the Free Software Foundation; either version 2
 * of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.

 * You should have received a copy of the GNU Lesser General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 */
package il.ac.biu.cs.grossmm.impl.activeData;

import il.ac.biu.cs.grossmm.api.data.Node;
import il.ac.biu.cs.grossmm.api.data.NodeType;
import il.ac.biu.cs.grossmm.api.flow.Converter;
import il.ac.biu.cs.grossmm.api.flow.NodeInspector;
import il.ac.biu.cs.grossmm.api.flow.NotificationProvider;
import il.ac.biu.cs.grossmm.api.flow.PublicationPointBase;
import il.ac.biu.cs.grossmm.api.flow.Status;
import il.ac.biu.cs.grossmm.api.flow.Subscriber;
import il.ac.biu.cs.grossmm.api.flow.Subscription;
import il.ac.biu.cs.grossmm.api.keys.Key;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Represents a data layer for some node type. A data layer for node type N consists
 * of all publication and notification points for node type N.
 *
 * @param <N> node type
 */
public class DataLayer<N> {

	final ActiveDataManagerImpl adm;

	/**
	 * Data layer handles data for a single data type represented by the
	 * following node type
	 */
	final NodeType<N> nodeType;

	/**
	 * List of all registered subscription points
	 */
	List<SubscriptionPointImpl> spList = new ArrayList<SubscriptionPointImpl>();

	/**
	 * List of all registered notifictaion providers. Publication points and
	 * notification points are notification providers.
	 */
	List<NotificationProvider> npList = new ArrayList<NotificationProvider>();

	/**
	 * Map of publication points indexed by their names
	 */
	Map<String, PublicationPointBase> ppMap = new HashMap<String, PublicationPointBase>();

	/**
	 * Map of all subscriptions on this layer indexed by their keys (note that
	 * both points of the subscription always belong to the same layer, event if
	 * one (or both of the points) is a converter)
	 */
	Map<Key, SubscriptionImpl<N>> subscriptions = new ConcurrentHashMap<Key, SubscriptionImpl<N>>();

	/**
	 * Map of best direct (non-chained) unconditional convertions from node type
	 * associated with this layer to node types of other layers
	 */
	Map<DataLayer, ConversionProcessor<N, ?>> conversionMap = new HashMap<DataLayer, ConversionProcessor<N, ?>>();

	/**
	 * Map of best unconditional conversion chains to convert node types of
	 * other layers to node type of this layer
	 */
	Map<DataLayer, ConversionChain> conversionChains = new HashMap<DataLayer, ConversionChain>();

	/**
	 * All conditional converters to convert node types from given layer to node
	 * types of this layer (grouped by source layer)
	 */
	private final Map<DataLayer, Set<ConversionProcessor>> conditionalsMap = new HashMap<DataLayer, Set<ConversionProcessor>>();

	Lock subscriptionsLock;

	/**
	 * Constructs data layer
	 * 
	 * @param adm
	 *            active data manager to conatin the data layer being created
	 * @param nodeType
	 *            node type associated with the data layer being created
	 */
	public DataLayer(ActiveDataManagerImpl adm, NodeType<N> nodeType) {
		this.adm = adm;
		this.nodeType = nodeType;

		subscriptionsLock = new ReentrantLock();
	}

	/*
	 * public void activated(Subscription<N> subscription, Root<N> root) {
	 * assert (subscription.root == null); // link node to subscription
	 * subscription.activate(root); // invoke activate method of all subscribers
	 * for this subscription for (SubscriptionPointImpl<N> s :
	 * subscription.subscriptionPoints) s.subscriber.activated(subscription.key,
	 * root); }
	 * 
	 * public void activated(Key key, Root<N> root) { Subscription<N>
	 * subscription = subscriptions.get(key);
	 * 
	 * if (subscription != null) activated(subscription, root); }
	 * 
	 * public void deactivated(Key key, SubscriptionState status) { Subscription<N>
	 * subscription = subscriptions.get(key); if (subscription == null) throw
	 * new RuntimeException("No such subscription");
	 * 
	 * subscription.deactivate(status);
	 * 
	 * for (SubscriptionPointImpl sp : subscription.subscriptionPoints)
	 * sp.subscriber.deactivated(key, status); }
	 */

	/**
	 * Registers a notification provider (either a notification point or
	 * publication point) on this layer
	 * 
	 * @param np
	 *            the notification provider to register
	 */
	void addNotificationProvider(NotificationProvider np) {
		for (int i = 0; i < npList.size(); i++) {
			NotificationProvider n = npList.get(i);
			if (n.getPriority() <= np.getPriority()) {
				npList.add(i, np);
				return;
			}
		}

		npList.add(np);
		return;
	}

	/**
	 * Deregisters a notification provider (either a notification point or
	 * publication point) on this layer
	 * 
	 * @param np
	 *            the notification provider to deregister
	 */
	public void removeNotificationProvider(NotificationProvider np) {
		npList.remove(np);
	}

	/**
	 * Registers a publication point on this layer
	 * 
	 * @param np
	 *            the publication point to register
	 */
	public void addSubscriptionPoint(SubscriptionPointImpl<N> sp) {
		spList.add(sp);
	}

	/**
	 * Deregisters a publication point on this layer
	 * 
	 * @param np
	 *            the publication point to deregister
	 */
	public void removeSubscriptionPoint(SubscriptionPointImpl<N> sp) {
		spList.remove(sp);
	}

	/**
	 * Given a key return a data node for this key on this layer
	 * 
	 * @param <T>
	 *            node type
	 * @param key
	 *            the key
	 * @return the data node for the specified key
	 */
	@SuppressWarnings("unchecked")
	<T> Root<T> getRoot(Key key) {
		// perform the operation atomically on subscriptions
		subscriptionsLock.lock();
		try {
			SubscriptionImpl subscription = subscriptions.get(key);

			// if no subscription for this key is registered -> return null
			if (subscription == null)
				return null;

			return subscription.getRoot();
		} finally {
			subscriptionsLock.unlock();
		}
	}

	/**
	 * Checks if there's already a subscription for a given key provided by some
	 * output point on this layer. If yes, atomically adds a subscriber to the
	 * list of subscribers on that subscription, and returns the subscription.
	 * 
	 * @param key
	 *            the key
	 * @param subscriber
	 *            the subscriber
	 * @return subscribtion if found (cache hit), otherwise null
	 */
	protected SubscriptionImpl<N> tryCached(Key key, Subscriber<N> subscriber) {
		subscriptionsLock.lock();

		try {

			SubscriptionImpl<N> s = subscriptions.get(key);

			if (s != null) {
				s.add(subscriber);

				Status status = s.getStatus();
				try {
					if (status.isKindOf(Status.ACTIVE))
						subscriber.activated(key, s.getRoot());
					else
						subscriber.deactivated(key, status);
				} catch (Exception e) {
					e.printStackTrace();
					return null;
				}
			}

			return s;
		} finally {
			subscriptionsLock.unlock();
		}
	}

	/**
	 * Registers a new subscription on this layer if no another subscription
	 * with the same key is registered on this layer. Returns the subscription
	 * registered for the specified key after the operation. The operation is
	 * performed atomically.
	 * 
	 * @param key
	 *            the subscription kay
	 * @param subscription
	 *            the <tt>Subscription</tt> object
	 * @param subscriber
	 *            the subscriber to receive notifications
	 * @return the subscription registered for the specified key after the
	 *         operation (i.e. the newly registered subscription if no
	 *         subscription was registered for this key, otherwise the
	 *         previously registered subscription
	 */
	protected SubscriptionImpl<N> registerSubscription(Key key,
			SubscriptionImpl<N> subscription, Subscriber<N> subscriber) {

		// do it atimically on subscriptions..
		subscriptionsLock.lock();

		try {

			SubscriptionImpl<N> s = subscriptions.get(key);

			if (s != null) {
				s.add(subscriber);
				// return existing subscription
				return s;
			}

			// register new subscription
			subscriptions.put(key, subscription);
			subscription.add(subscriber);

			return subscription;

		} finally {
			subscriptionsLock.unlock();
		}
	}

	/**
	 * Unsubscribes the given subscriber from a resource with the given key
	 * 
	 * @param key
	 *            the key
	 * @param subscriber
	 *            the subscriber
	 * @return true iff the subscriber was subscribed to the resource
	 */
	protected boolean unsubscribe(Key key, Subscriber<N> subscriber) {
		subscriptionsLock.lock();

		try {

			SubscriptionImpl<N> s = subscriptions.get(key);

			if (s != null) {
				return s.remove(subscriber);
			}

			// no such subsription...
			return false;

		} finally {
			subscriptionsLock.unlock();
		}
	}

	/**
	 * Gets node inspector for the given node
	 * 
	 * @param node
	 *            the node
	 * @return node inspector for the given node
	 */
	public NodeInspector getInspector(Node node) {
		return adm.getInspector(node.getClass());
	}

	/**
	 * Registeres a new conversion processor on this layer 0
	 * 
	 * @param <Ndst>
	 *            the destination node type for this conversion processor
	 * @param conversion
	 *            the conversion processor
	 * @param dstLayer
	 *            the destination layer for this conversino processor
	 */
	protected synchronized <Ndst> void register(
			ConversionProcessor<N, Ndst> conversion, DataLayer<Ndst> dstLayer) {
		if (conversion.isConditional()) {
			Set<ConversionProcessor> set = conditionalsMap.get(dstLayer);
			if (set == null) {
				set = new HashSet<ConversionProcessor>();
				dstLayer.conditionalsMap.put(this, set);
			}

			set.add(conversion);
		} else
			conversionMap.put(dstLayer, conversion);
	}

	/**
	 * Given a destination data layer, returns a conversion processor which
	 * converts active data from this layer to the specified destination layer,
	 * or null if no such conversion processor registered.
	 * 
	 * @param to
	 *            the sestination data layer
	 * @return a conversion processor which converts active data from this layer
	 *         to the specified destination layer, or null if no such conversion
	 *         processor registered.
	 */
	protected Converter<N, ?> getConverterTo(DataLayer to) {
		return conversionMap.get(to);
	}

	/**
	 * Sets best conversion chains from this data layer to all data layers,
	 * reachable from this data layer
	 * 
	 * @param bestChains
	 *            best conversion chains from this data layer to all data
	 *            layers, reachable from this data layer
	 */
	protected void setConversionChains(
			Map<DataLayer, ConversionChain> bestChains) {
		conversionChains = bestChains;

		updateLinks();
	}

	/**
	 * Updates dispatchers for all subscription points
	 */
	protected void updateLinks() {
		// update dispatchers for all subscription points
		for (SubscriptionPointImpl<N> sp : spList)
			sp.dispatcher.updateLinks();
	}

	/**
	 * Gets list of all notification providers (pubication and notifiation
	 * points) registered on this layer.
	 * 
	 * @return list of all notification providers registered on this layer
	 */
	public List<NotificationProvider> getNotificationProviders() {
		return npList;
	}

	/**
	 * Gets set of all layers conditionally reachable from this layer.
	 * 
	 * @return set of all layers conditionally reachable from this layer
	 */
	public Set<DataLayer> getConditionallyReachableLayers() {
		return conditionalsMap.keySet();
	}

	/**
	 * 
	 * Gets a set of all conditional converters that convert nodes on the given
	 * layer to node types of this layer (grouped by source layer).
	 * 
	 * @param fromLayer
	 *            the source data layer
	 * @return set of all conditional converters that convert nodes on the given
	 *         layer to node types of this layer (grouped by source layer)
	 */
	public Set<ConversionProcessor> getConditionals(DataLayer fromLayer) {
		return conditionalsMap.get(fromLayer);
	}

	/**
	 * Deregisteres a subscription. Should be invoked when a subscription has
	 * been closed because no more subscribers use it.
	 * 
	 * @param subscription
	 *            the subscription which has been closed.
	 */
	protected void subscriptionClosed(Subscription<N> subscription) {
		subscriptions.remove(subscription);
	}

	/**
	 * Gets a suibscrion for a given key, and locks it is such subscription
	 * exists. If no such subscription exists, returns null.
	 * <p>
	 * A subscription cannot be deregistered while it is in locked state.
	 * At most one subscription must be in a locked state at a time.
	 * 
	 * @param key
	 *            the subscription key
	 * @return the subscription for the given key
	 */
	public SubscriptionImpl<N> lockSubscription(Key key) {
		SubscriptionImpl<N> s;
		subscriptionsLock.lock();
		s = subscriptions.get(key);
		if (s == null)
			subscriptionsLock.unlock();
		return s;
	}

	/**
	 * Unlocks a previously locked subscription.
	 * @param subscription the subscription to be unlocked
	 */
	public void unlockSubscription(SubscriptionImpl<N> subscription) {
		subscriptionsLock.unlock();
	}

	@Override
	public String toString() {
		return "layer for " + nodeType.toString();
	}

}
